package com.reactim.message.routing;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.reactim.message.connection.RSocketRequesterHolder;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.messaging.rsocket.RSocketRequester;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Mono;

import jakarta.annotation.PostConstruct;
import jakarta.annotation.PreDestroy;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;

/**
 * 基于Kafka的消息路由服务实现
 */
@Service
@Slf4j
@RequiredArgsConstructor
public class KafkaMessageRoutingService implements MessageRoutingService {
    
    private final KafkaTemplate<String, String> kafkaTemplate;
    private final RSocketRequesterHolder requesterHolder;
    private final ObjectMapper objectMapper;
    
    @Value("${spring.application.name:reactim-message-service}")
    private String applicationName;
    
    @Value("${server.port:8082}")
    private String serverPort;
    
    @Value("${message.routing.kafka.topic:cross-instance-messages}")
    private String crossInstanceTopic;
    
    @Value("${message.routing.kafka.ack-topic:message-acknowledgments}")
    private String ackTopic;
    
    @Value("${message.routing.retry.max-attempts:3}")
    private int maxRetryAttempts;
    
    @Value("${message.routing.ack.timeout-seconds:30}")
    private long ackTimeoutSeconds;
    
    private String instanceId;
    private volatile boolean subscriptionActive = false;
    
    // 存储等待确认的消息
    private final ConcurrentHashMap<String, PendingMessage> pendingMessages = new ConcurrentHashMap<>();
    
    @PostConstruct
    public void init() {
        // 生成唯一的实例ID
        this.instanceId = applicationName + "-" + serverPort + "-" + UUID.randomUUID().toString().substring(0, 8);
        log.info("消息路由服务初始化完成，实例ID: {}", instanceId);
        
        // 启动消息订阅
        subscribeCrossInstanceMessages();
    }
    
    @PreDestroy
    public void destroy() {
        stopSubscription();
    }
    
    @Override
    public <T> Mono<Boolean> publishCrossInstanceMessage(String targetInstanceId, Long userId, String route, T data) {
        return Mono.fromCallable(() -> {
            try {
                // 创建跨实例消息
                CrossInstanceMessage message = CrossInstanceMessage.builder()
                        .messageId(UUID.randomUUID().toString())
                        .sourceInstanceId(instanceId)
                        .targetInstanceId(targetInstanceId)
                        .userId(userId)
                        .route(route)
                        .data(objectMapper.writeValueAsString(data))
                        .messageType(data.getClass().getSimpleName())
                        .timestamp(System.currentTimeMillis())
                        .retryCount(0)
                        .requiresAck(true)
                        .build();
                
                // 序列化消息
                String messageJson = objectMapper.writeValueAsString(message);
                
                // 发送到Kafka
                CompletableFuture<SendResult<String, String>> future = kafkaTemplate.send(
                    crossInstanceTopic, 
                    targetInstanceId, 
                    messageJson
                );
                
                // 等待发送结果
                SendResult<String, String> result = future.get(5, TimeUnit.SECONDS);
                
                // 如果需要确认，添加到待确认列表
                if (message.getRequiresAck()) {
                    PendingMessage pendingMessage = new PendingMessage(message, System.currentTimeMillis());
                    pendingMessages.put(message.getMessageId(), pendingMessage);
                    
                    // 设置超时清理
                    scheduleAckTimeout(message.getMessageId());
                }
                
                log.debug("跨实例消息发送成功: messageId={}, targetInstance={}, userId={}", 
                    message.getMessageId(), targetInstanceId, userId);
                
                return true;
                
            } catch (Exception e) {
                log.error("发送跨实例消息失败: targetInstance={}, userId={}, error={}", 
                    targetInstanceId, userId, e.getMessage());
                return false;
            }
        })
        .onErrorReturn(false);
    }
    
    @Override
    public void subscribeCrossInstanceMessages() {
        subscriptionActive = true;
        log.info("开始订阅跨实例消息，实例ID: {}", instanceId);
    }
    
    @Override
    public void stopSubscription() {
        subscriptionActive = false;
        log.info("停止订阅跨实例消息，实例ID: {}", instanceId);
    }
    
    @Override
    public Mono<Boolean> isAvailable() {
        return Mono.fromCallable(() -> {
            try {
                // 简单的健康检查：尝试发送一个测试消息到Kafka
                kafkaTemplate.send(crossInstanceTopic + "-health", instanceId, "ping")
                    .get(2, TimeUnit.SECONDS);
                return true;
            } catch (Exception e) {
                log.warn("消息路由服务健康检查失败: {}", e.getMessage());
                return false;
            }
        })
        .onErrorReturn(false);
    }
    
    /**
     * 监听跨实例消息
     */
    @KafkaListener(topics = "${message.routing.kafka.topic:cross-instance-messages}")
    public void handleCrossInstanceMessage(ConsumerRecord<String, String> record) {
        if (!subscriptionActive) {
            return;
        }
        
        String targetInstanceId = record.key();
        String messageJson = record.value();
        
        // 检查消息是否是发给当前实例的
        if (!instanceId.equals(targetInstanceId)) {
            log.debug("收到非本实例的消息，忽略: targetInstance={}, currentInstance={}", 
                targetInstanceId, instanceId);
            return;
        }
        
        try {
            // 反序列化消息
            CrossInstanceMessage message = objectMapper.readValue(messageJson, CrossInstanceMessage.class);
            
            log.debug("收到跨实例消息: messageId={}, sourceInstance={}, userId={}, route={}", 
                message.getMessageId(), message.getSourceInstanceId(), message.getUserId(), message.getRoute());
            
            // 处理消息
            handleIncomingMessage(message)
                .doOnSuccess(success -> {
                    // 发送确认
                    if (message.getRequiresAck()) {
                        sendAcknowledgment(message.getMessageId(), message.getSourceInstanceId(), 
                            success ? MessageAcknowledgment.AckStatus.SUCCESS : MessageAcknowledgment.AckStatus.FAILED, 
                            null);
                    }
                })
                .doOnError(error -> {
                    log.error("处理跨实例消息失败: messageId={}, error={}", message.getMessageId(), error.getMessage());
                    // 发送失败确认
                    if (message.getRequiresAck()) {
                        sendAcknowledgment(message.getMessageId(), message.getSourceInstanceId(), 
                            MessageAcknowledgment.AckStatus.FAILED, error.getMessage());
                    }
                })
                .subscribe();
                
        } catch (Exception e) {
            log.error("解析跨实例消息失败: messageJson={}, error={}", messageJson, e.getMessage());
        }
    }
    
    /**
     * 监听消息确认
     */
    @KafkaListener(topics = "${message.routing.kafka.ack-topic:message-acknowledgments}")
    public void handleMessageAcknowledgment(ConsumerRecord<String, String> record) {
        if (!subscriptionActive) {
            return;
        }
        
        String targetInstanceId = record.key();
        String ackJson = record.value();
        
        // 检查确认是否是发给当前实例的
        if (!instanceId.equals(targetInstanceId)) {
            return;
        }
        
        try {
            MessageAcknowledgment ack = objectMapper.readValue(ackJson, MessageAcknowledgment.class);
            
            log.debug("收到消息确认: messageId={}, status={}", ack.getMessageId(), ack.getStatus());
            
            // 从待确认列表中移除
            PendingMessage pendingMessage = pendingMessages.remove(ack.getMessageId());
            if (pendingMessage != null) {
                if (ack.getStatus() == MessageAcknowledgment.AckStatus.SUCCESS) {
                    log.debug("消息确认成功: messageId={}", ack.getMessageId());
                } else {
                    log.warn("消息处理失败: messageId={}, error={}", ack.getMessageId(), ack.getErrorMessage());
                    // 可以在这里实现重试逻辑
                    handleMessageFailure(pendingMessage.getMessage(), ack.getErrorMessage());
                }
            }
            
        } catch (Exception e) {
            log.error("解析消息确认失败: ackJson={}, error={}", ackJson, e.getMessage());
        }
    }
    
    /**
     * 处理接收到的跨实例消息
     */
    private Mono<Boolean> handleIncomingMessage(CrossInstanceMessage message) {
        return Mono.fromCallable(() -> {
            try {
                // 根据用户ID获取本地的RSocketRequester
                // 这里需要从本地缓存中查找用户的连接
                // 由于我们无法直接访问ConnectionManager，我们需要通过RSocketRequesterHolder来处理
                
                // 首先尝试找到用户的连接
                RSocketRequester requester = findUserRequester(message.getUserId());
                if (requester == null) {
                    log.warn("用户连接不存在于当前实例: userId={}, instanceId={}", 
                        message.getUserId(), instanceId);
                    return false;
                }
                
                // 反序列化消息数据
                Object messageData = deserializeMessageData(message.getData(), message.getMessageType());
                
                // 发送消息给用户
                requester.route(message.getRoute())
                    .data(messageData)
                    .send()
                    .block(); // 这里使用block()是因为我们在Callable中
                
                log.debug("跨实例消息发送给用户成功: userId={}, route={}", 
                    message.getUserId(), message.getRoute());
                
                return true;
                
            } catch (Exception e) {
                log.error("处理跨实例消息失败: messageId={}, error={}", message.getMessageId(), e.getMessage());
                return false;
            }
        });
    }
    
    /**
     * 查找用户的RSocketRequester
     */
    private RSocketRequester findUserRequester(Long userId) {
        return requesterHolder.getRequesterByUserId(userId);
    }
    
    /**
     * 反序列化消息数据
     */
    private Object deserializeMessageData(String data, String messageType) throws JsonProcessingException {
        // 根据消息类型反序列化数据
        // 对于跨实例消息路由，我们通常处理的是通用的消息对象
        // 这里简化处理，直接返回Map对象，让RSocket客户端处理具体的反序列化
        return objectMapper.readValue(data, java.util.Map.class);
    }
    
    /**
     * 发送消息确认
     */
    private void sendAcknowledgment(String messageId, String targetInstanceId, 
                                   MessageAcknowledgment.AckStatus status, String errorMessage) {
        try {
            MessageAcknowledgment ack = MessageAcknowledgment.builder()
                    .messageId(messageId)
                    .instanceId(instanceId)
                    .status(status)
                    .timestamp(System.currentTimeMillis())
                    .errorMessage(errorMessage)
                    .build();
            
            String ackJson = objectMapper.writeValueAsString(ack);
            
            kafkaTemplate.send(ackTopic, targetInstanceId, ackJson);
            
            log.debug("发送消息确认: messageId={}, status={}, targetInstance={}", 
                messageId, status, targetInstanceId);
                
        } catch (Exception e) {
            log.error("发送消息确认失败: messageId={}, error={}", messageId, e.getMessage());
        }
    }
    
    /**
     * 处理消息失败
     */
    private void handleMessageFailure(CrossInstanceMessage message, String errorMessage) {
        if (message.getRetryCount() < maxRetryAttempts) {
            // 增加重试次数
            message.setRetryCount(message.getRetryCount() + 1);
            
            log.info("重试发送跨实例消息: messageId={}, retryCount={}", 
                message.getMessageId(), message.getRetryCount());
            
            // 重新发送消息
            publishCrossInstanceMessage(message.getTargetInstanceId(), message.getUserId(), 
                message.getRoute(), message.getData())
                .subscribe();
        } else {
            log.error("跨实例消息重试次数超限，放弃发送: messageId={}, maxRetries={}", 
                message.getMessageId(), maxRetryAttempts);
        }
    }
    
    /**
     * 设置确认超时
     */
    private void scheduleAckTimeout(String messageId) {
        // 使用简单的延时任务来处理超时
        Mono.delay(java.time.Duration.ofSeconds(ackTimeoutSeconds))
            .subscribe(tick -> {
                PendingMessage pendingMessage = pendingMessages.remove(messageId);
                if (pendingMessage != null) {
                    log.warn("消息确认超时: messageId={}", messageId);
                    handleMessageFailure(pendingMessage.getMessage(), "Acknowledgment timeout");
                }
            });
    }
    
    /**
     * 待确认消息
     */
    private static class PendingMessage {
        private final CrossInstanceMessage message;
        private final long timestamp;
        
        public PendingMessage(CrossInstanceMessage message, long timestamp) {
            this.message = message;
            this.timestamp = timestamp;
        }
        
        public CrossInstanceMessage getMessage() {
            return message;
        }
        
        public long getTimestamp() {
            return timestamp;
        }
    }
}